/**
 * collectd - src/rrdcached.c
 * Copyright (C) 2008  Florian octo Forster
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; only version 2 of the License is applicable.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
 *
 * Authors:
 *   Florian octo Forster <octo at verplant.org>
 **/
/**
   This is a plugin that uses mysql as storage backend!
**/

#include "collectd.h"
#include "plugin.h"
#include "common.h"
#include "configfile.h"

#include <time.h>
#ifdef HAVE_MYSQL_H
#include <mysql.h>
#elif defined(HAVE_MYSQL_MYSQL_H)
#include <mysql/mysql.h>
#endif

struct mysql_database_s /* {{{ */
{
  /* instance == NULL  =>  legacy mode */
  char *instance;
  char *host;
  char *user;
  char *pass;
  char *database;
  char *socket;
  int   port;

  MYSQL *con;
  int    state;
};
typedef struct mysql_database_s mysql_database_t; /* }}} */

static mysql_database_t *mydb;
static char schema[2048];
static const int MAX_QUERY_LEN = 4096;
static char query_str[MAX_QUERY_LEN];

static void mysql_database_free (void *arg) /* {{{ */
{
  mysql_database_t *db;

  DEBUG ("mysql plugin: mysql_database_free (arg = %p);", arg);

  db = (mysql_database_t *) arg;

  if (db == NULL)
    return;

  if (db->con != NULL)
    mysql_close (db->con);

  sfree (db->host);
  sfree (db->user);
  sfree (db->pass);
  sfree (db->socket);
  sfree (db->instance);
  sfree (db->database);
  sfree (db);
} /* }}} void mysql_database_free */

/* Configuration handling functions {{{
 *
 * <Plugin mysql>
 *   <Database "plugin_instance1">
 *     Host "localhost"
 *     Port 22000
 *     ...
 *   </Database>
 * </Plugin>
 }}} */

static int mysql_config_set_string (char **ret_string, /* {{{ */
  oconfig_item_t *ci)
{
  char *string;

  if ((ci->values_num != 1)
      || (ci->values[0].type != OCONFIG_TYPE_STRING))
  {
    WARNING ("mysql plugin: The `%s' config option "
             "needs exactly one string argument.", ci->key);
    return (-1);
  }

  string = strdup (ci->values[0].value.string);
  if (string == NULL)
  {
    ERROR ("mysql plugin: strdup failed.");
    return (-1);
  }

  if (*ret_string != NULL)
    free (*ret_string);
  *ret_string = string;

  return (0);
} /* }}} int mysql_config_set_string */

static int mysql_config_set_int (int *ret_int, /* {{{ */
  oconfig_item_t *ci)
{
  if ((ci->values_num != 1)
    || (ci->values[0].type != OCONFIG_TYPE_NUMBER))
  {
    WARNING ("mysql plugin: The `%s' config option "
             "needs exactly one string argument.", ci->key);
    return (-1);
  }

  *ret_int = ci->values[0].value.number;

  return (0);
} /* }}} int mysql_config_set_int */

static int mysql_config_set_boolean (int *ret_boolean, /* {{{ */
  oconfig_item_t *ci)
{
  int status = 0;

  if (ci->values_num != 1)
    status = -1;

  if (status == 0)
  {
    if (ci->values[0].type == OCONFIG_TYPE_BOOLEAN)
      *ret_boolean = ci->values[0].value.boolean;
    else if (ci->values[0].type == OCONFIG_TYPE_STRING)
    {
       if (IS_TRUE (ci->values[0].value.string))
         *ret_boolean = 1;
       else if (IS_FALSE (ci->values[0].value.string))
         *ret_boolean = 0;
       else
         status = -1;
    }
    else
      status = -1;
  }

  if (status != 0)
  {
    WARNING ("mysql plugin: The `%s' config option "
             "needs exactly one boolean argument.", ci->key);
    return (-1);
  }
  return (0);
} /* }}} mysql_config_set_boolean */

static int mysql_config (oconfig_item_t *ci) /* {{{ */
{
  mysql_database_t *db;
  int plugin_block;
  int status = 0;
  int i;

  if ((ci->values_num != 1)
      || (ci->values[0].type != OCONFIG_TYPE_STRING))
  {
    WARNING ("mysql plugin: The `Database' block "
       "needs exactly one string argument.");
    return (-1);
  }

  db = (mysql_database_t *) malloc (sizeof (*db));
  if (db == NULL)
  {
    ERROR ("mysql plugin: malloc failed.");
    return (-1);
  }
  memset (db, 0, sizeof (*db));

  /* initialize all the pointers */
  db->host     = NULL;
  db->user     = NULL;
  db->pass     = NULL;
  db->database = NULL;
  db->socket   = NULL;
  db->con      = NULL;

  plugin_block = 1;
  if (strcasecmp ("Plugin", ci->key) == 0)
  {
    db->instance = NULL;
  }
  else if (strcasecmp ("Database", ci->key) == 0)
  {
    plugin_block = 0;
    status = mysql_config_set_string (&db->instance, ci);
    if (status != 0)
    {
      sfree (db);
      return (status);
    }
    assert (db->instance != NULL);
  }
  else
  {
    ERROR ("mysql plugin: mysql_config: "
           "Invalid key: %s", ci->key);
    return (-1);
  }

  /* Fill the `mysql_database_t' structure.. */
  for (i = 0; i < ci->children_num; i++)
  {
    oconfig_item_t *child = ci->children + i;

    if (strcasecmp ("Host", child->key) == 0)
      status = mysql_config_set_string (&db->host, child);
    else if (strcasecmp ("User", child->key) == 0)
      status = mysql_config_set_string (&db->user, child);
    else if (strcasecmp ("Password", child->key) == 0)
      status = mysql_config_set_string (&db->pass, child);
    else if (strcasecmp ("Port", child->key) == 0)
      status = mysql_config_set_int (&db->port, child);
    else if (strcasecmp ("Socket", child->key) == 0)
      status = mysql_config_set_string (&db->socket, child);
    /* Check if we're currently handling the `Plugin' block. If so,
     * handle `Database' _blocks_, too. */
    else if ((plugin_block != 0)
      && (strcasecmp ("Database", child->key) == 0)
      && (child->children != NULL))
    {
      /* If `plugin_block > 1', there has been at least one
       * `Database' block */
      plugin_block++;
      status = mysql_config (child);
    }
    /* Now handle ordinary `Database' options (without children) */
    else if ((strcasecmp ("Database", child->key) == 0)
      && (child->children == NULL))
      status = mysql_config_set_string (&db->database, child);
    else
    {
      WARNING ("mysql plugin: Option `%s' not allowed here.", child->key);
      status = -1;
    }

    if (status != 0)
      break;
  }

  /* Check if there were any `Database' blocks. */
  if (plugin_block > 1)
  {
    /* There were connection blocks. Don't use any legacy stuff. */
    if ((db->host != NULL)
      || (db->user != NULL)
      || (db->pass != NULL)
      || (db->database != NULL)
      || (db->socket != NULL)
      || (db->port != 0))
    {
      WARNING ("mysql plugin: At least one <Database> "
               "block has been found. The legacy "
               "configuration will be ignored.");
    }
    mysql_database_free (db);
    return (0);
  }
  else if (plugin_block != 0)
  {
    WARNING ("mysql plugin: You're using the legacy "
             "configuration options. Please consider "
             "updating your configuration!");
  }

  /* Check that all necessary options have been given. */
  while (status == 0)
  {
    /* Zero is allowed and automatically handled by
     * `mysql_real_connect'. */
    if ((db->port < 0) || (db->port > 65535))
    {
      ERROR ("mysql plugin: Database %s: Port number out "
             "of range: %i",
             (db->instance != NULL)
              ? db->instance
              : "<legacy>",
              db->port);
      status = -1;
    }
    break;
  } /* while (status == 0) */

  /* If all went well, register this database for reading */
  if (status == 0)
  {
/*    user_data_t ud;
    char cb_name[DATA_MAX_NAME_LEN];

    DEBUG ("mysql plugin: Registering new read callback: %s",
      	(db->database != NULL) ? db->database : "<default>");

    memset (&ud, 0, sizeof (ud));
    ud.data = (void *) db;
    ud.free_func = mysql_database_free;

    if (db->database != NULL)
      ssnprintf (cb_name, sizeof (cb_name), "mysql-%s",
      		db->database);
    else
      sstrncpy (cb_name, "mysql", sizeof (cb_name));

    plugin_register_complex_read (NULL, cb_name,
      		      mysql_read,
      		      NULL, &ud);
*/
          mydb = db;
  }
  else
  {
    mysql_database_free (db);
    return (-1);
  }

  return (0);
} /* }}} int mysql_config*/

static int value_list_to_string (char *buffer, int buffer_len,/*{{{*/
    const data_set_t *ds, const value_list_t *vl)
{
  int offset;
  int status;
  int i;

  assert (0 == strcmp (ds->type, vl->type));

  memset (buffer, '\0', buffer_len);

  status = ssnprintf (buffer, buffer_len, "%u", (unsigned int) vl->time);
  if ((status < 1) || (status >= buffer_len))
    return (-1);
  offset = status;

  for (i = 0; i < ds->ds_num; i++)
  {
    if ((ds->ds[i].type != DS_TYPE_COUNTER)
        && (ds->ds[i].type != DS_TYPE_GAUGE)
        && (ds->ds[i].type != DS_TYPE_DERIVE)
        && (ds->ds[i].type != DS_TYPE_ABSOLUTE))
      return (-1);

    if (ds->ds[i].type == DS_TYPE_COUNTER)
    {
      status = ssnprintf (buffer + offset, buffer_len - offset,
          ":%llu", vl->values[i].counter);
    }
    else if (ds->ds[i].type == DS_TYPE_GAUGE)
    {
      status = ssnprintf (buffer + offset, buffer_len - offset,
          ":%f", vl->values[i].gauge);
    }
    else if (ds->ds[i].type == DS_TYPE_DERIVE) {
      status = ssnprintf (buffer + offset, buffer_len - offset,
    ":%"PRIi64, vl->values[i].derive);
    }
    else /* if (ds->ds[i].type == DS_TYPE_ABSOLUTE) */ {
      status = ssnprintf (buffer + offset, buffer_len - offset,
    ":%"PRIu64, vl->values[i].absolute);
    }

    if ((status < 1) || (status >= (buffer_len - offset)))
      return (-1);

    offset += status;
  } /* for ds->ds_num */

  return (0);
} /* int value_list_to_string *//*}}}*/

static int value_list_to_tablename (char *buffer, int buffer_len,/*{{{*/
    const data_set_t *ds, const value_list_t *vl)
{
  int offset = 0;
  int status;

  assert (0 == strcmp (ds->type, vl->type));
/*
  if (datadir != NULL)
  {
    status = ssnprintf (buffer + offset, buffer_len - offset,
        "%s/", datadir);
    if ((status < 1) || (status >= buffer_len - offset))
      return (-1);
    offset += status;
  }

  status = ssnprintf (buffer + offset, buffer_len - offset,
      "%s/", vl->host);
  if ((status < 1) || (status >= buffer_len - offset))
    return (-1);
  offset += status;
*/

  if (strcmp(vl->plugin, "mrjob") == 0 &&
      strlen(vl->plugin_instance) > 0) {
    status = ssnprintf (buffer + offset, buffer_len - offset,
                    "mrjob-%s", vl->plugin_instance);
    if ((status < 1) || (status >= buffer_len - offset))
            return (-1);
    offset += status;
    if (strlen (vl->type_instance) > 0)
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%s", vl->type_instance);
    else
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%s", vl->type);
  } else {
    if (strlen (vl->plugin_instance) > 0)
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%s-%s-", vl->plugin, vl->plugin_instance);
    else
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%s-", vl->plugin);
    if ((status < 1) || (status >= buffer_len - offset))
      return (-1);
    offset += status;

    if (strlen (vl->type_instance) > 0)
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%s-%s", vl->type, vl->type_instance);
    else
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%s", vl->type);
    if ((status < 1) || (status >= buffer_len - offset))
      return (-1);
    offset += status;
  }

/*
  strncpy (buffer + offset, ".rrd", buffer_len - offset);
*/
  buffer[buffer_len - 1] = 0;

  return (0);
} /* int value_list_to_filename *//*}}}*/

static MYSQL *getconnection (mysql_database_t *db)
{/*{{{*/
  if (db->state != 0)
  {
    int err;
    if ((err = mysql_ping (db->con)) != 0)
    {
      WARNING ("mysql_ping failed for %s: %s",
      		(db->instance != NULL)
      		? db->instance
      		: "<legacy>",
      		mysql_error (db->con));
      db->state = 0;
    }
    else
    {
      db->state = 1;
      return (db->con);
    }
  }

  if ((db->con = mysql_init (db->con)) == NULL)
  {
    ERROR ("mysql_init failed: %s", mysql_error (db->con));
    db->state = 0;
    return (NULL);
  }

  if (mysql_real_connect (db->con, db->host, db->user, db->pass,
      db->database, db->port, db->socket, 0) == NULL)
  {
    ERROR ("mysql plugin: Failed to connect to database %s "
           "at server %s: %s",
           (db->database != NULL) ? db->database : "<none>",
           (db->host != NULL) ? db->host : "localhost",
           mysql_error (db->con));
    db->state = 0;
    return (NULL);
  }
  else
  {
    INFO ("mysql plugin: Successfully connected to database %s "
      	"at server %s (server version: %s, protocol version: %d)",
      	(db->database != NULL) ? db->database : "<none>",
      	mysql_get_host_info (db->con),
      	mysql_get_server_info (db->con),
      	mysql_get_proto_info (db->con));
    db->state = 1;
    return (db->con);
  }
} /* static MYSQL *getconnection (mysql_database_t *db) *//*}}}*/

/*
static void submit (const char *type, const char *type_instance,
    value_t *values, size_t values_len, mysql_database_t *db)
{
  value_list_t vl = VALUE_LIST_INIT;

  vl.values     = values;
  vl.values_len = values_len;

  set_host (db, vl.host, sizeof (vl.host));

  sstrncpy (vl.plugin, "mysql", sizeof (vl.plugin));
  set_plugin_instance (db, vl.plugin_instance, sizeof (vl.plugin_instance));

  sstrncpy (vl.type, type, sizeof (vl.type));
  if (type_instance != NULL)
    sstrncpy (vl.type_instance, type_instance, sizeof (vl.type_instance));

  plugin_dispatch_values (&vl);
}
static void set_host (mysql_database_t *db, char *buf, size_t buflen)
{
  if (db->instance == NULL)
    sstrncpy (buf, hostname_g, buflen);
  else
  {
    if ((db->host == NULL)
      	|| (strcmp ("", db->host) == 0)
      	|| (strcmp ("localhost", db->host) == 0))
      sstrncpy (buf, hostname_g, buflen);
    else
      sstrncpy (buf, db->host, buflen);
  }
}
*/

static MYSQL_RES *exec_query (MYSQL *con, const char *query)
{/*{{{*/
  MYSQL_RES *res;

  int query_len = strlen (query);

  if (mysql_real_query (con, query, query_len))
  {
    ERROR ("mysql plugin: Failed to execute query: %s",
      	mysql_error (con));
    INFO ("mysql plugin: SQL query was: %s", query);
    return (NULL);
  }

  res = mysql_store_result (con);
  if (res == NULL)
  {
    ERROR ("mysql plugin: Failed to store query result: %s",
      	mysql_error (con));
    INFO ("mysql plugin: SQL query was: %s", query);
    return (NULL);
  }

  return (res);
} /* exec_query *//*}}}*/

static int wrap_ssnprintf_int(char *buffer, size_t buffer_len, int *offset,
                          const char *arg, const int val)
{/*{{{*/
  int status;
  status = ssnprintf (buffer + *offset, buffer_len - *offset, arg, val);
  if ((status < 1) || (status >= buffer_len - *offset))
          return (-1);
  *offset += status;
  return 0;
}/*}}}*/

static int wrap_ssnprintf_str(char *buffer, size_t buffer_len, int *offset,
                          const char *arg, const char* val)
{/*{{{*/
  int status;
  status = ssnprintf (buffer + *offset, buffer_len - *offset, arg, val);
  if ((status < 1) || (status >= buffer_len - *offset))
          return (-1);
  *offset += status;
  return 0;
}/*}}}*/

static int generate_schema(char *buf, size_t buflen)
{/*{{{*/
  int i;
  int offset = 0;
  for (i = 1; i <= 64; ++i) {
    if (wrap_ssnprintf_int(buf, buflen, &offset, "cloud%d DOUBLE", i) < 0)
      return -1;
  }
  return 0;
}/*}}}*/

static int create_table(char *tablename, mysql_database_t *db)
{/*{{{*/
  int offset = 0;
  MYSQL_RES *res;
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "CREATE TABLE IF NOT EXISTS %s", tablename);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "(time TIMESTAMP", NULL);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "%s", schema);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     ")", NULL);
  res = exec_query (db->con, query_str);
  return 0;
}/*}}}*/

static int mysql_insert(char *tablename, time_t time, const char *host,
  const char *value, mysql_database_t *db)
{
  int offset = 0;
  MYSQL_RES *res;
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "INSERT INTO %s", tablename);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "(time, %s)", host);
  char buf[80];
  struct tm *ts;
  ts = localtime(&time);
  strftime(buf, sizeof(buf), "%Y-%m-%d %H:%M:%S", ts);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "value (%s, ", buf);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "%s)", value);
  res = exec_query (db->con, query_str);
  return 0;
}

static int mysql_init (void)
{/*{{{*/
  generate_schema(schema, sizeof(schema));
  return (0);
} /* int rc_init *//*}}}*/

static int mysql_collectd_write (const data_set_t *ds, const value_list_t *vl,
    user_data_t __attribute__((unused)) *user_data)
{
  char tablename[512];
  char values[512];
  int status;

  if (mydb->con == NULL)
  {
    ERROR ("rrdcached plugin: MYDB->CON == NULL.");
    plugin_unregister_write ("rrdcached");
    return (-1);
  }

  if (strcmp (ds->type, vl->type) != 0)
  {
    ERROR ("rrdcached plugin: DS type does not match value list type");
    return (-1);
  }

  if (value_list_to_tablename (tablename, sizeof (tablename), ds, vl) != 0)
  {
    ERROR ("rrdcached plugin: value_list_to_filename failed.");
    return (-1);
  }

  if (value_list_to_string (values, sizeof (values), ds, vl) != 0)
  {
    ERROR ("rrdcached plugin: value_list_to_string failed.");
    return (-1);
  }

  status = create_table(tablename, mydb);
  if (status != 0)
  {
    ERROR ("rrdcached plugin: create (%s) failed", tablename);
    return (-1);
  }

  status = mysql_insert(tablename, vl->time, vl->host, values, mydb);
  if (status != 0)
  {
    ERROR ("rrdcached plugin: mysql_update (%s, [%s], 1) failed",
        tablename, values, status);
    return (-1);
  }

  return (0);
} /* int rc_write */

static int mysql_collectd_shutdown (void)
{
  mysql_database_free(mydb);
  return (0);
} /* int rc_shutdown */

void module_register (void)
{
  plugin_register_config ("mysql", mysql_config);
  plugin_register_init ("mysql", mysql_collectd_init);
  plugin_register_write ("mysql", mysql_collectd_write, /* user_data = */ NULL);
  plugin_register_shutdown ("mysql", mysql_collectd_shutdown);
} /* void module_register */

